package com.pingan.haofang.searchcloud.quartz.service.impl;

import com.google.common.collect.Lists;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.pingan.haofang.searchcloud.common.dto.OperationResult;
import com.pingan.haofang.searchcloud.index.constants.IndexBuildConstants;
import com.pingan.haofang.searchcloud.index.constants.IndexBuildStatus;
import com.pingan.haofang.searchcloud.index.constants.IndexDataSourceType;
import com.pingan.haofang.searchcloud.index.dao.IndexDao;
import com.pingan.haofang.searchcloud.index.domain.Index;
import com.pingan.haofang.searchcloud.index.dto.IndexBuildProgressDTO;
import com.pingan.haofang.searchcloud.index.dto.IndexExecutionDTO;
import com.pingan.haofang.searchcloud.index.service.IndexBuildProgressService;
import com.pingan.haofang.searchcloud.index.service.IndexBuildService;
import com.pingan.haofang.searchcloud.indexdatasource.dao.IndexDatasourceDao;
import com.pingan.haofang.searchcloud.indexdatasource.dto.IndexDatasourceDTO;
import com.pingan.haofang.searchcloud.indexdatasource.dto.ScheduleTaskDto;
import com.pingan.haofang.searchcloud.indexdatasource.facade.IndexDatasourceScheduleFacade;
import com.pingan.haofang.searchcloud.indexdatasource.service.IndexDatasourceService;
import com.pingan.haofang.searchcloud.quartz.constants.QuartzConstants;
import com.pingan.haofang.searchcloud.quartz.dto.ArgsDto;
import com.pingan.haofang.searchcloud.quartz.dto.ScheduleTask;
import com.pingan.haofang.searchcloud.quartz.dto.TaskQueryDto;
import com.pingan.haofang.searchcloud.quartz.service.IndexBuildScheduleTaskService;
import com.pingan.haofang.searchcloud.quartz.service.ScheduleTaskManager;
import com.pingan.haofang.searchcloud.user.dao.SearchProjectDao;
import com.pingan.haofang.searchcloud.user.domain.SearchProject;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.quartz.JobDetail;
import org.quartz.JobKey;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.TriggerKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author ex-sunbixiao001
 * @date 2018/6/12
 * @since 1.0.0
 */
@Service
public class IndexBuildScheduleTaskServiceImpl implements IndexBuildScheduleTaskService {
    static Logger LOG = LoggerFactory.getLogger(IndexBuildScheduleTaskServiceImpl.class);

    private static Gson GSON = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss").create();

    @Autowired
    private ScheduleTaskManager scheduleTaskManager;

    @Autowired
    SearchProjectDao projectDao;

    @Autowired
    IndexDatasourceService indexDatasourceService;

    @Autowired
    IndexDatasourceDao datasourceDao;

    @Autowired
    IndexDao indexDao;

    @Autowired
    protected StringRedisTemplate redisTemplate;

    @Autowired
    private IndexBuildService indexBuildService;

    @Autowired
    private IndexBuildProgressService indexBuildProgressService;

    @Autowired
    private IndexDatasourceScheduleFacade indexDatasourceScheduleFacade;

    @Value("${batch.log.url:}")
    private String logurl;

    @Autowired
    private Scheduler scheduler;

    @Override
    public void refreshAll() throws SchedulerException {
        //获得所有有效任务
        List<ScheduleTaskDto> scheduleTaskDtos = indexDatasourceScheduleFacade.getAllEffectiveTasks(new TaskQueryDto());
        //注册
        register(scheduleTaskDtos, QuartzConstants.TARGET, QuartzConstants.FULL_METHOD);
        Set<TriggerKey> newKeys = scheduleTaskDtos.stream()
                .map(item -> TriggerKey.triggerKey(item.getIndexName(), item.getProjectCode()))
                .collect(Collectors.toSet());
        Collection deletingKeys = CollectionUtils.removeAll(scheduleTaskManager.getAllFull().keySet(), newKeys);

        for (Object key : deletingKeys) {
            TriggerKey triggerKey = (TriggerKey) key;
            scheduleTaskManager.delete(triggerKey.getGroup(), triggerKey.getName());
        }
    }

    @Override
    public void refresh(Long indexId) throws SchedulerException {
        //根据indexId获得索引相关的信息
        TaskQueryDto dto = new TaskQueryDto();
        dto.setIndexId(indexId);
        List<ScheduleTaskDto> scheduleTaskDtos = indexDatasourceScheduleFacade.getAllEffectiveTasks(dto);
        ScheduleTaskDto taskDto = null;

        //有任务
        if (CollectionUtils.isNotEmpty(scheduleTaskDtos)) {
            taskDto = scheduleTaskDtos.get(0);
            //重新注册
            register(Lists.newArrayList(taskDto), QuartzConstants.TARGET, QuartzConstants.FULL_METHOD);
        } else {
            Index index = indexDao.findOne(indexId);
            SearchProject project = projectDao.findOne(index.getProjectId());

            //验证是否有job ，有job则删除
            JobDetail jobDetail = scheduleTaskManager.get(project.getProjectCode(), index.getIndexName());
            if (jobDetail != null) {
                scheduleTaskManager.delete(project.getProjectCode(), index.getIndexName());
            }
        }
    }

    /**
     * 注册、更新全量 task
     *
     * @param scheduleTaskDtos
     * @param target
     * @param method
     */
    private void register(List<ScheduleTaskDto> scheduleTaskDtos, String target, String method) {

        List<ScheduleTask> result = new ArrayList<>();
        for (ScheduleTaskDto dto : scheduleTaskDtos) {

            ScheduleTask scheduleTask = new ScheduleTask();
            scheduleTask.setGroup(dto.getProjectCode());
            scheduleTask.setName(dto.getIndexName());
            scheduleTask.setDesc(dto.getIndexName() + "_" + dto.getProjectCode());
            scheduleTask.setTarget(target);
            ArgsDto argsDto = new ArgsDto();
            argsDto.setProgressId(0L);
            argsDto.setIndexName(dto.getIndexName());
            argsDto.setIndexId(dto.getIndexId());
            argsDto.setProjectCode(dto.getProjectCode());
            argsDto.setIndexDatasourceList(dto.getIndexDatasourceDTOs());
            scheduleTask.setArgs(argsDto);
            scheduleTask.setMethod(method);
            scheduleTask.setCron(dto.getCron());
            scheduleTaskManager.register(scheduleTask);
        }
    }


    /**
     * 手动调度
     *
     * @param message
     */
    @Override
    public void doProcess(IndexExecutionDTO message) {

        // 获取最新的数据源信息
        List<IndexDatasourceDTO> indexDatasourceDTOList = null;
        if (StringUtils.isNotBlank(message.getDataSourceName())) {
            IndexDatasourceDTO indexDatasourceDTO = indexDatasourceService.findDatasource(
                    message.getProjectCode(), message.getIndexName(), message.getDataSourceName());
            indexDatasourceDTOList = Arrays.asList(indexDatasourceDTO);
        } else {
            indexDatasourceDTOList = indexDatasourceService.findDatasource(message.getProjectCode(),
                    message.getIndexName());
        }

        ArgsDto argsDto = new ArgsDto();
        argsDto.setProgressId(message.getProgressId());
        argsDto.setIndexDatasourceList(indexDatasourceDTOList);

        // 触发任务调度
        scheduleTaskManager.run(message.getProjectCode(), message.getIndexName(), argsDto);

    }

    /**
     * 中斷任務
     *
     * @param id
     */
    @Override
    public void interrupt(Long id) {
        LOG.info("job will be interrupted,progressId:{}", id);
        redisTemplate.opsForHash().put(IndexBuildConstants.INTERRUPT, id + "", IndexBuildConstants.INTERRUPT);

    }

    /**
     * 暂停任務
     *
     * @param progressId
     */
    @Override
    public OperationResult pause(Long progressId) {

        String taskStr = (String) redisTemplate.opsForHash().get(IndexBuildConstants.SOLR_PROGRESS_ID_PREFIX +
                        progressId,
                IndexBuildConstants.SCHEDULE_TASK);

        if (taskStr == null) {
            return OperationResult.of(false, "根据该progressId:" + progressId + "没有找到可以暂停的任务");
        }

        LOG.info("job will be pause,progressId:{}", progressId);

        redisTemplate.opsForHash().put(IndexBuildConstants.PAUSE, Long.toString(progressId), taskStr);

        return OperationResult.of(true, "暂停任务成功！");

    }

    @Transactional
    @Override
    public OperationResult resume(Long progressId) {

        String taskStr = (String) redisTemplate.opsForHash().get(IndexBuildConstants.PAUSE, Long.toString(progressId));

        if (taskStr == null) {
            return OperationResult.of(false, "根据该progressId:" + progressId + "没有找到可以恢复的任务");
        }
        LOG.info("job will be resume,progressId:{}", progressId);

        ScheduleTask task = GSON.fromJson(taskStr, ScheduleTask.class);

        //删除暂停标记
        redisTemplate.opsForHash().delete(IndexBuildConstants.PAUSE, Long.toString(progressId));


        IndexBuildProgressDTO progressDTO = indexBuildProgressService.query(progressId);

        progressDTO.setFailReason(null);

        //设置为进行中
        progressDTO.setStatus(IndexBuildStatus.RUNNING.getValue());

        indexBuildProgressService.updateProgress(progressDTO);


        //恢复任务
        scheduleTaskManager.resume(task.getGroup(), task.getName());

        return OperationResult.of(true, "正在恢复任务！");
    }

    @Override
    public OperationResult abort(Long progressId) {

        String taskStr = (String) redisTemplate.opsForHash().get(IndexBuildConstants.PAUSE, Long.toString(progressId));
        if (taskStr == null) {
            return OperationResult.of(false, "根据该progressId:" + progressId + "没有找到可以放弃的任务");
        }

        LOG.info("job will be abort,progressId:{}", progressId);

        ScheduleTask task = GSON.fromJson(taskStr, ScheduleTask.class);

        //删除暂停标记
        redisTemplate.opsForHash().delete(IndexBuildConstants.PAUSE, Long.toString(progressId));

        List<IndexDatasourceDTO> indexDatasourceList = task.getArgs().getIndexDatasourceList();

        //删除 redis中的进度信息
        indexBuildService.deleteProgressAfterIndexBuild(indexDatasourceList, progressId);

        IndexBuildProgressDTO progressDTO = indexBuildProgressService.query(progressId);

        //设置为终止
        progressDTO.setStatus(IndexBuildStatus.STOP.getValue());

        indexBuildProgressService.updateProgress(progressDTO);


        //解锁
        indexBuildService.unLock(task.getGroup(), task.getName());

        return OperationResult.of(true, "放弃任务成功！");
    }



    /**
     * 增量定时任务初始化
     */
    @ManagedOperation
    @Override
    public void incScheduleInit() throws SchedulerException {
        LOG.info("inc schedule init start");
        List<ScheduleTaskDto> scheduleTaskDtos = indexDatasourceScheduleFacade.getAllEffectiveTasks(new TaskQueryDto());
        Set<TriggerKey> newKeys = new HashSet<>();
        //遍历每个索引
        for (ScheduleTaskDto dto : scheduleTaskDtos) {
            // 遍历索引每个数据源，过滤出来带增量的
            if (org.springframework.util.StringUtils.isEmpty(dto.getIncCron())) {
                scheduleTaskManager.delete(dto.getProjectCode() + QuartzConstants.INC_SCHEDULE_GROUP_SUFFIX, dto.getIndexName());
                continue;
            }
            boolean hasInc = false;
            for (IndexDatasourceDTO datasourceDTO : dto.getIndexDatasourceDTOs()) {
                //是否有增量
                if (IndexDataSourceType.NOT_SURPPORT.getValue() != datasourceDTO.getIncDatasourceType()
                        && null != datasourceDTO.getIncDatasourceRpc()
                        ) {

                    ScheduleTask scheduleIncTask = new ScheduleTask();
                    scheduleIncTask.setGroup(dto.getProjectCode() + QuartzConstants.INC_SCHEDULE_GROUP_SUFFIX);
                    scheduleIncTask.setName(dto.getIndexName());
                    scheduleIncTask.setDesc(dto.getProjectCode() + "_" + dto.getIndexName());
                    scheduleIncTask.setTarget(QuartzConstants.TARGET);

                    ArgsDto argsDto = new ArgsDto();
                    argsDto.setProgressId(0L);
                    argsDto.setIndexName(dto.getIndexName());
                    argsDto.setIndexId(dto.getIndexId());
                    argsDto.setProjectCode(dto.getProjectCode());
                    argsDto.setIndexDatasourceList(dto.getIndexDatasourceDTOs());
                    scheduleIncTask.setArgs(argsDto);
                    scheduleIncTask.setMethod(QuartzConstants.INC_METHOD);
                    scheduleIncTask.setCron(dto.getIncCron());
                    newKeys.add(TriggerKey.triggerKey(dto.getIndexName(), dto.getProjectCode() + QuartzConstants.INC_SCHEDULE_GROUP_SUFFIX));
                    scheduleTaskManager.register(scheduleIncTask);
                    LOG.info("inc schedule init success,projectCode:{},indexName:{},cron:{}", dto.getProjectCode(),
                            dto.getIndexName(), dto.getIncCron());
                    hasInc = true;
                    break;
                }
            }
            //如果没有增量，就删除旧的增量schedule
            if (!hasInc) {
                scheduleTaskManager.delete(dto.getProjectCode() + QuartzConstants.INC_SCHEDULE_GROUP_SUFFIX, dto.getIndexName());
            }
        }
        LOG.info("inc schedule init finished ");
    }

    /**
     * 刷新单个索引的增量任务
     *
     * @param indexId
     */
    @ManagedOperation
    @Override
    public void incScheduleRefresh(Long indexId) {
        Objects.requireNonNull(indexId, "index id can not be null");
        List<ScheduleTaskDto> scheduleTaskDtos = indexDatasourceScheduleFacade.getAllEffectiveTasks(new TaskQueryDto() {{
            this.setIndexId(indexId);
        }});
        for (ScheduleTaskDto dto : scheduleTaskDtos) {
            // 遍历索引每个数据源，过滤出来带增量的
            if (org.springframework.util.StringUtils.isEmpty(dto.getIncCron())) {
                scheduleTaskManager.delete(dto.getProjectCode() + QuartzConstants.INC_SCHEDULE_GROUP_SUFFIX, dto.getIndexName());
                continue;
            }
            boolean hasInc = false;
            for (IndexDatasourceDTO datasourceDTO : dto.getIndexDatasourceDTOs()) {
                //是否有增量
                if (IndexDataSourceType.NOT_SURPPORT.getValue() != datasourceDTO.getIncDatasourceType()
                        && null != datasourceDTO.getIncDatasourceRpc()
                        ) {
                    ScheduleTask scheduleIncTask = new ScheduleTask();
                    scheduleIncTask.setGroup(dto.getProjectCode() + QuartzConstants.INC_SCHEDULE_GROUP_SUFFIX);
                    scheduleIncTask.setName(dto.getIndexName());
                    scheduleIncTask.setDesc(dto.getProjectCode() + "_" + dto.getIndexName());
                    scheduleIncTask.setTarget(QuartzConstants.TARGET);

                    ArgsDto argsDto = new ArgsDto();
                    argsDto.setProgressId(0L);
                    argsDto.setIndexName(dto.getIndexName());
                    argsDto.setIndexId(dto.getIndexId());
                    argsDto.setProjectCode(dto.getProjectCode());
                    argsDto.setIndexDatasourceList(dto.getIndexDatasourceDTOs());
                    scheduleIncTask.setArgs(argsDto);
                    scheduleIncTask.setMethod(QuartzConstants.INC_METHOD);
                    scheduleIncTask.setCron(dto.getIncCron());
                    scheduleTaskManager.register(scheduleIncTask);
                    LOG.info("inc schedule inited,projectCode:{},indexName:{},cron:{}", dto.getProjectCode(), dto.getIndexName(), dto.getIncCron());
                    hasInc = true;
                    break;
                }
            }
            //如果没有增量，就删除旧的增量schedule
            if (!hasInc) {
                scheduleTaskManager.delete(dto.getProjectCode() + QuartzConstants.INC_SCHEDULE_GROUP_SUFFIX, dto.getIndexName());
            }
        }
        LOG.info("inc schedule refreshed ，index id:{}", indexId);
    }

    @ManagedOperation
    @Override
    public void incSchedulePause(Long indexId) {
        if (Objects.isNull(indexId)) {
            return;
        }
        List<ScheduleTaskDto> scheduleTaskDtos = indexDatasourceScheduleFacade.getAllEffectiveTasks(new TaskQueryDto() {{
            this.setIndexId(indexId);
        }});
        for (ScheduleTaskDto scheduleTaskDto : scheduleTaskDtos) {
            JobKey jobKey = new JobKey(scheduleTaskDto.getIndexName(), scheduleTaskDto.getProjectCode() + QuartzConstants.INC_SCHEDULE_GROUP_SUFFIX);
            try {
                if (scheduler.checkExists(jobKey)) {
                    scheduleTaskManager.pause(jobKey);
                }
            } catch (SchedulerException e) {
                LOG.error(String.format("inc schedule pause error, invalid schedule group:%s,name:%s", jobKey.getGroup(), jobKey.getName()), e);
                e.printStackTrace();
            }
        }
    }

    @ManagedOperation
    @Override
    public void incScheduleResume(Long indexId) {
        if (Objects.isNull(indexId)) {
            return;
        }
        List<ScheduleTaskDto> scheduleTaskDtos = indexDatasourceScheduleFacade.getAllEffectiveTasks(new TaskQueryDto() {{
            this.setIndexId(indexId);
        }});
        for (ScheduleTaskDto scheduleTaskDto : scheduleTaskDtos) {
            JobKey jobKey = new JobKey(scheduleTaskDto.getIndexName(), scheduleTaskDto.getProjectCode() + QuartzConstants.INC_SCHEDULE_GROUP_SUFFIX);
            try {
                if (scheduler.checkExists(jobKey)) {
                    scheduler.resumeJob(jobKey);
                }
            } catch (SchedulerException e) {
                LOG.error(String.format("inc schedule resume error, invalid schedule group:%s,name:%s", jobKey.getGroup(), jobKey.getName()), e);
                e.printStackTrace();
            }
        }
    }

    @Override
    public String getLogUrl(Long progressId) {
        return String.format(logurl, progressId);
    }

}
